package com.yuaer.demo.springsparkjobrunnerdemo.runner;

import com.yuaer.demo.springsparkjobrunnerdemo.feign.ActPlanService;
import com.yuaer.demo.springsparkjobrunnerdemo.feign.MirsvrDemoService;
import com.yuaer.demo.springsparkjobrunnerdemo.model.MktActivity;
import lombok.extern.slf4j.Slf4j;
import org.apache.spark.launcher.SparkAppHandle;
import org.apache.spark.launcher.SparkLauncher;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.time.LocalDate;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 正式活动执行器
 */
@Slf4j
@Component
public class NormalActivityRunner extends ScriptRunner {

    @Autowired
    private ActPlanService actPlanService;

    @Autowired
    private MirsvrDemoService mirsvrDemoService;

    //从nacos中获取并行度
    @Value("${spark.job.parallel:10}")
    private int parallelJobThreads;

    //参数名称, 这里命名规范参考linux命令行格式
    public static final String ARG_NAME_EXCUTE_DATE = "excute-date";
    public static final String ARG_NAME_ACTIVITY_NO = "activity-no";

    //todo 这部分可以考虑放到配置中
    public static final String SPARK_JOB_PARAM_NAME_PREFIX = "activity_";
    public static final String SPARK_JOB_PARAM_JAR_PATH = "/app/ccmms/etl/lib/xxx.jar";
    public static final String SPARK_JOB_PARAM_MAIN_CLASS = "com.yuzer.sparkjob.ActivityReacher";
    public static final String SPARK_JOB_PARAM_HOME = "~/spark";
    public static final String SPARK_JOB_PARAM_DEPLOY_MODE = "1";

    /**
     * 通过构造函数定义任务名称
     */
    public NormalActivityRunner() {
        super("normal");
    }

    @Override
    public void runTask(ApplicationArguments args) throws IllegalAccessException {

        String excuteDateStr = getArg(args, ARG_NAME_EXCUTE_DATE, null);
        if (excuteDateStr == null) {
            throw new IllegalAccessException("未找到参数: " + ARG_NAME_EXCUTE_DATE);
        }
        LocalDate excuteDate = LocalDate.parse(excuteDateStr);
        String activityNo = getArg(args, ARG_NAME_ACTIVITY_NO, null);
//        List<MktActivity> activities = actPlanService.getActivitiesToBeExecute(excuteDate, activityNo);
        List<MktActivity> activities = mirsvrDemoService.getActivitiesToBeExecute();

        if (activities.isEmpty()) {
            log.info("未找到待跑活动");
            return;
        }

        //并行执行spark job
        ExecutorService executor = Executors.newFixedThreadPool(parallelJobThreads);    //创建线程池 并发线程数从nacos中获取
        //收集跑批结果
        CompletableFuture.allOf(activities.stream()
                        .map(act -> CompletableFuture.supplyAsync(() -> executeSparkJob(act), executor))
                        .toArray(CompletableFuture[]::new))
                        .whenComplete((res, ex) -> {
                            if (ex != null) {
                                log.error("活动跑批失败", ex);
                            } else {
                                //todo 更新任务跑批状态

                            }
                        });
        executor.shutdown();

    }

    private boolean executeSparkJob(MktActivity activity) {
        System.out.println("执行调度任务:" + activity);
        //1. 创建SparkLauncher和配置参数
        SparkLauncher sparkLauncher = new SparkLauncher()
                .setAppName(SPARK_JOB_PARAM_NAME_PREFIX + activity.getMoveNo() + "_" + activity.getType())
                .setAppResource(SPARK_JOB_PARAM_JAR_PATH)
                .setMainClass(SPARK_JOB_PARAM_MAIN_CLASS)
                .setMaster("") //以下配置省略
                .setSparkHome(SPARK_JOB_PARAM_HOME)
                .setDeployMode(SPARK_JOB_PARAM_DEPLOY_MODE)
                //以下配置省略
                .setConf("spark.executor.memory", "1g")
                .setConf("spark.executor.cores", "1")
                .setConf("spark.yarn.queue", "default")
                .setConf("spark.driver.memory", "1g");

        //这里增加任务需要的参数
        sparkLauncher.addAppArgs("ActivityReacher.jar",
                "--inputCurrentDate=" + activity.getStartDate(),
                "--inputMoveTypes=" + activity.getTestFlag(),
                "--inputMoveNos=" + activity.getMoveNo()
                );
        //这个参数问下联通
        sparkLauncher.addFile("");

        //2. 启作业和监听
        SparkAppHandle handle;
        try {
            handle = sparkLauncher.setVerbose(true).startApplication(new SparkAppHandle.Listener() {
                @Override
                public void stateChanged(SparkAppHandle sparkAppHandle) {
                    log.info("任务: {} 状态变更, {}", sparkAppHandle.getAppId(), sparkAppHandle.getState().toString());
                }

                @Override
                public void infoChanged(SparkAppHandle sparkAppHandle) {
                    log.info("任务: {} 执行信息, {}", sparkAppHandle.getAppId(), sparkAppHandle.getState().toString());

                }
            });
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

        //3. 等待作业完成
        while (!handle.getState().isFinal()) {
            try {
                Thread.sleep(5000); // 每5秒检查一次状态
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return false;
            }
        }
        return handle.getState().isFinal();
    }
}